package com.gmail.dengtao.joe.group.transport.impl.filter;

import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gmail.dengtao.joe.commons.util.ByteUtils;
import com.gmail.dengtao.joe.group.packet.DefaultPacketImpl;
import com.gmail.dengtao.joe.group.protocol.Protocol;
import com.gmail.dengtao.joe.group.protocol.ProtocolFactory;
import com.gmail.dengtao.joe.transport.filter.FilterAdapter;
import com.gmail.dengtao.joe.transport.filter.FilterEntity;
import com.gmail.dengtao.joe.transport.session.Session;

/**
 * Assembling the packets into a protocol.
 * @author <a href="mailto:joe.dengtao@gmail.com">DengTao</a>
 * @version 1.0
 * @since 1.0
 */
public class ProtocolFilter extends FilterAdapter {
	
	private Logger LOGGER = LoggerFactory.getLogger(ProtocolFilter.class);

	private Map<Session, Map<String, PacketBuffer>> packetCacheMap = new HashMap<Session, Map<String, PacketBuffer>>();
	
	private ProtocolFactory protocoFactory;
	
	public ProtocolFilter(ProtocolFactory protocoFactory) {
		this.protocoFactory = protocoFactory;
	}
	
	@Override
	public void dataReceived(FilterEntity nextEntity, Session session,
			Object data) throws Exception {
		DefaultPacketImpl packet = (DefaultPacketImpl) data;
		if (packet.getSerialNumber() == 0 && !packet.hasMorePacket()) {	// Not Subcontract Packet...
			// Parse Packet to packet. 
			Protocol protocol = protocoFactory.parse(packet.getType(), packet.getData());
			// Push to next...
			super.dataReceived(nextEntity, session, protocol);
		} else { // Subcontract Packet...
			Map<String, PacketBuffer> packetBuffers = null;
			if (packetCacheMap.containsKey(session)) {
				packetBuffers = packetCacheMap.get(session);
			} else {
				packetBuffers = new HashMap<String, PacketBuffer>();
				packetCacheMap.put(session, packetBuffers);
			}
			PacketBuffer packetBuffer = null;
			if (packetBuffers.containsKey(packet.getPacketNumber() + "")) {
				packetBuffer = packetBuffers.get(packet.getPacketNumber() + "");
			} else {
				packetBuffer = new PacketBuffer(packet.getType(), packet.getPacketNumber());
				packetBuffers.put(packet.getPacketNumber() + "", packetBuffer);
			}
			try {
				packetBuffer.addPacket(packet);
			} catch (Exception e) {
				LOGGER.warn("Add packet Error:", e);
			} finally {
				if (packetBuffer.isReady()) {
					Protocol protocol = protocoFactory.parse(packetBuffer.getType(), packetBuffer.toBytes());
					// Push to next...
					super.dataReceived(nextEntity, session, protocol);
				} else {
					LOGGER.warn("Packets:" + packetBuffer.getPn() + ",SN=" + packet.getSerialNumber() + ", Not Ready!");
				}
			}
		}
		
		//
	}

	@Override
	public void sessionClosed(FilterEntity nextEntity, Session session)
			throws Exception {
		packetCacheMap.remove(session);
		super.sessionClosed(nextEntity, session);
	}

	class PacketBuffer {
		
		private int pn;
		private int type;

		private boolean ready = false;
		private int maxSn = 0;
		
		private Map<String, DefaultPacketImpl> packets = new HashMap<String, DefaultPacketImpl>();
		
		public PacketBuffer(int type, int pn) {
			this.pn = pn;
			this.type = type;
		}

		public int getType() {
			return type;
		}

		public void addPacket(DefaultPacketImpl packet) {
			if (ready) {
				throw new IllegalStateException();
			}
			packets.put(packet.getSerialNumber() + "", packet);
			maxSn = packet.getSerialNumber() > maxSn ? packet.getSerialNumber() : maxSn;
			//System.out.println(maxSn + ", " + packets.size() + "," + packet.getSerialNumber() + ", " + packets.get(maxSn + ""));
			if (maxSn == packets.size() - 1 && packets.get(maxSn + "").getMorePacket() == 0) {
				ready = true;
			}
		}
		
		public int getPn() {
			return pn;
		}

		public byte[] toBytes() {
			byte[] result = ByteUtils.EMPTY_BYTE_ARRAY;
			for (int i = 0; i < maxSn; i++) {
				DefaultPacketImpl packet = packets.get(i + "");
				result = ByteUtils.merge(result, packet.getData());
			}
			return result;
		}
		
		public boolean isReady() {
			return ready;
		}
	}
	
}
